/*
 * Copyright (c) 2020 - present, Inspur Genersoft Co., Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.iec.edp.caf.rpc.remote.grpc;

import com.google.protobuf.ByteString;
import io.grpc.Metadata;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
import io.iec.edp.caf.commons.core.SerializerFactory;
import io.iec.edp.caf.commons.core.enums.SerializeType;
import io.iec.edp.caf.commons.exception.CAFRuntimeException;
import io.iec.edp.caf.commons.exception.ExceptionLevel;
import io.iec.edp.caf.commons.exception.entity.DefaultExceptionProperties;
import io.iec.edp.caf.commons.exception.entity.ExceptionErrorCode;
import io.iec.edp.caf.commons.utils.SpringBeanUtils;
import io.iec.edp.caf.commons.utils.StringUtils;
import io.iec.edp.caf.rpc.api.common.GspSerializeType;
import io.iec.edp.caf.rpc.api.entity.RpcContext;
import io.iec.edp.caf.rpc.api.event.RpcServerEventBroker;
import io.iec.edp.caf.rpc.api.grpc.GrpcExceptionMetaData;
import io.iec.edp.caf.rpc.api.grpc.GrpcInvokeServiceGrpc;
import io.iec.edp.caf.rpc.api.grpc.GrpcRequest;
import io.iec.edp.caf.rpc.api.grpc.GrpcResponse;
import io.iec.edp.caf.rpc.api.serialize.RpcSerializeUtil;
import io.iec.edp.caf.rpc.api.service.InternalServiceManageService;
import io.iec.edp.caf.rpc.api.service.RpcServer;
import io.iec.edp.caf.rpc.api.spi.RpcContextResolver;
import io.iec.edp.caf.rpc.api.spi.RpcContextTransmittee;
import io.iec.edp.caf.rpc.api.support.ConstanceVarible;
import io.iec.edp.caf.rpc.api.support.RpcThreadCacheHolder;
import io.iec.edp.caf.rpc.api.utils.Utils;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.apache.commons.lang3.exception.ExceptionUtils;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.time.OffsetDateTime;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.UUID;

import static io.grpc.stub.ServerCalls.asyncUnimplementedStreamingCall;
import static io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall;

@Slf4j
public class GrpcServerInvoke extends GrpcInvokeServiceGrpc.GrpcInvokeServiceImplBase {

    private RpcServer rpcServer = SpringBeanUtils.getBean(RpcServer.class);
    private RpcServerEventBroker serverEventBroker  = SpringBeanUtils.getBean(RpcServerEventBroker.class);

    @SneakyThrows
    @Override
    public void grpcRemoteInvoke(GrpcRequest request, StreamObserver<GrpcResponse> responseObserver) {
        String serviceId = request.getServiceId();
        LinkedHashMap<String, Object> params = new LinkedHashMap<>(request.getParamsMap());
        String version = request.getVersion();

        String serviceUnitName ;
        String tenantIdString ;
        Integer tenantId = -1 ;
        String result;
        var localDict = new HashMap<String, Object>();
        var eventContextDict = new HashMap<String, String>();
        RpcContext rpcContext = RpcContext.getInstance();
        try {
            //get gsp context from filter
            var context = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_CONTEXT);
            Map<String,String> gspContext = (HashMap<String, String>) SerializerFactory.getDeserializer(SerializeType.Json).deserialize(context, HashMap.class);

            if (!gspContext.containsKey(ConstanceVarible.GSP_RPC)
                    || gspContext.get(ConstanceVarible.GSP_RPC) == null
                    || gspContext.get(ConstanceVarible.GSP_RPC).length() == 0
                    || !"true".equalsIgnoreCase(gspContext.get(ConstanceVarible.GSP_RPC))){
                responseObserver.onError(Status.PERMISSION_DENIED.withDescription("Only allow invoke in iGIX rpc framework").asException());
                throw new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT,
                        DefaultExceptionProperties.RESOURCE_FILE,
                        ExceptionErrorCode.illegalRPCCall,
                        new String[]{},
                        null, ExceptionLevel.Error, false);
            }


            //var contextId = gspContext.get("ContextId");
            serviceUnitName = gspContext.get(ConstanceVarible.GSP_MSU);
            tenantIdString = gspContext.get(ConstanceVarible.GSP_RPC_TENANT);
            tenantId = !StringUtils.isEmpty(tenantIdString) ? Integer.valueOf(tenantIdString) : null;

            //RpcContext: prepare tenant and msu
            rpcContext.putContext("transmitte-msu",serviceUnitName);
            rpcContext.putContext("transmitte-tenantid",tenantId);
            RpcContextResolver.replay(rpcContext,
                    RpcContextTransmittee.TransmitteType.TENANT.getValue(),
                    RpcContextTransmittee.TransmitteType.MSU.getValue());

            gspContext.remove(ConstanceVarible.GSP_RPC);
            gspContext.remove(ConstanceVarible.GSP_CONTEXT_ID);
            gspContext.remove(ConstanceVarible.GSP_MSU);

            var eventContext = gspContext.get(ConstanceVarible.GSP_RPC_CLIENT_ENVENT);
            eventContextDict = SerializerFactory.getDeserializer(SerializeType.Json).deserialize(eventContext, HashMap.class);
            localDict.put(ConstanceVarible.LOG_MSU, serviceUnitName);
            this.serverEventBroker.firePreRpcInvokeEvent(eventContextDict, params, localDict);

            var intermanager = SpringBeanUtils.getBean(InternalServiceManageService.class);
            var rpcServiceMethodDefinition = intermanager.getRpcMethodDefinition(serviceId);
            var parameters = RpcSerializeUtil.deSerializeParameter4RPC2(rpcServiceMethodDefinition,params);
//            var parameters = RpcSerializeUtil.deSerializeParameter(rpcServiceMethodDefinition,params);
            result = rpcServer.invokeService(serviceId, "", parameters, GspSerializeType.Protobuf).toString();
        } catch (Throwable e) {

            var statusRuntimeException = handleException(e);
            responseObserver.onError(statusRuntimeException);
            serverEventBroker.fireExceptionRpcInvokeEvent(eventContextDict, params, localDict, statusRuntimeException);
            throw e;
        } finally {
            RpcContextResolver.restore(rpcContext,RpcContextTransmittee.TransmitteType.TENANT.getValue());
//            restoreTenant(tenantId);
        }


        var contextDict = new HashMap<String, String>();
        this.serverEventBroker.firePostRpcInvokeEvent(contextDict, result, localDict);
        var eventContextStr = SerializerFactory.getSerializer(SerializeType.Json).serializeToString(contextDict);
        RpcThreadCacheHolder.setValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT, eventContextStr);

        GrpcResponse resp = GrpcResponse.newBuilder()
                .setMessage(result)
                .build();
        responseObserver.onNext(resp);
        responseObserver.onCompleted();
    }

    @SneakyThrows
    @Override
    public void grpcRemoteInvokeFromStream(GrpcRequest request,
                                           StreamObserver<GrpcResponse> responseObserver) {
        String serviceId = request.getServiceId();
        LinkedHashMap<String, Object> params = new LinkedHashMap<>(request.getParamsMap());
        String version = request.getVersion();

        //get gsp context from filter
        var context = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_CONTEXT);
        Map<String,String> gspContext = (HashMap<String, String>) SerializerFactory.getDeserializer(SerializeType.Json).deserialize(context, HashMap.class);

        if (!gspContext.containsKey(ConstanceVarible.GSP_RPC)
                || gspContext.get(ConstanceVarible.GSP_RPC) == null
                || gspContext.get(ConstanceVarible.GSP_RPC).length() == 0
                || !"true".equalsIgnoreCase(gspContext.get(ConstanceVarible.GSP_RPC))){
            var state = handleException(Status.PERMISSION_DENIED.withDescription("Only allow invoke in iGIX rpc framework").asException());
            handleException(state);

            throw new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT,
                    DefaultExceptionProperties.RESOURCE_FILE,
                    ExceptionErrorCode.illegalRPCCall,
                    new String[]{},
                    null, ExceptionLevel.Error, false);
        }


        //var contextId = gspContext.get("ContextId");
        String serviceUnitName = gspContext.get(ConstanceVarible.GSP_MSU);
        String tenantIdString = gspContext.get(ConstanceVarible.GSP_RPC_TENANT);
        Integer tenantId = !StringUtils.isEmpty(tenantIdString) ? Integer.valueOf(tenantIdString) : null;

        //RpcContext: prepare tenant and msu
        RpcContext rpcContext = RpcContext.getInstance();
        rpcContext.putContext("transmitte-msu",serviceUnitName);
        rpcContext.putContext("transmitte-tenantid",tenantId);
        RpcContextResolver.replay(rpcContext,
                RpcContextTransmittee.TransmitteType.TENANT.getValue(),
                RpcContextTransmittee.TransmitteType.MSU.getValue());

        gspContext.remove(ConstanceVarible.GSP_RPC);
        gspContext.remove(ConstanceVarible.GSP_CONTEXT_ID);
        gspContext.remove(ConstanceVarible.GSP_MSU);

        var eventContext = gspContext.get(ConstanceVarible.GSP_RPC_CLIENT_ENVENT);
        var eventContextDict = SerializerFactory.getDeserializer(SerializeType.Json).deserialize(eventContext, HashMap.class);
        var localDict = new HashMap<String, Object>();
        localDict.put(ConstanceVarible.LOG_MSU, serviceUnitName);
        this.serverEventBroker.firePreRpcInvokeEvent(eventContextDict, params, localDict);

        InputStream result = null;
        try {
            var intermanager = SpringBeanUtils.getBean(InternalServiceManageService.class);
            var rpcServiceMethodDefinition = intermanager.getRpcMethodDefinition(serviceId);
            var parameters = RpcSerializeUtil.deSerializeParameter4RPC2(rpcServiceMethodDefinition,params);
//            var parameters = RpcSerializeUtil.deSerializeParameter(rpcServiceMethodDefinition,params);
            result = rpcServer.invokeServiceStream(serviceId, "", parameters);
            // download file as chunk
            byte[] bytes = new byte[4096];
            int size;
            while ((size = result.read(bytes)) > 0){
                var content = ByteString.copyFrom(bytes, 0 , size);
                GrpcResponse resp = GrpcResponse.newBuilder()
                        .setMessage("")
                        .setContent(content)
                        .build();
                responseObserver.onNext(resp);
            }
        } catch (Throwable e) {

            var statusRuntimeException = handleException(e);
            responseObserver.onError(statusRuntimeException);
            serverEventBroker.fireExceptionRpcInvokeEvent(eventContextDict, params, localDict, statusRuntimeException);
            throw e;
        } finally {
            RpcContextResolver.restore(rpcContext,RpcContextTransmittee.TransmitteType.TENANT.getValue());
//            restoreTenant(tenantId);
            if(result!=null){
                result.close();
            }
        }

        var contextDict = new HashMap<String, String>();
        this.serverEventBroker.firePostRpcInvokeEvent(contextDict, result, localDict);
        var eventContextStr = SerializerFactory.getSerializer(SerializeType.Json).serializeToString(contextDict);
        RpcThreadCacheHolder.setValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT, eventContextStr);

        responseObserver.onCompleted();
    }

    @SneakyThrows
    @Override
    public StreamObserver<GrpcRequest> grpcRemoteInvokeToStream(
            StreamObserver<GrpcResponse> responseObserver) {
        return new StreamObserver<GrpcRequest>() {
            String result = "";
            HashMap<String,String> eventContextDict;
            LinkedHashMap<String, Object> params;
            HashMap<String, Object> localDict;
            ByteArrayInputStream swapStream;
            @SneakyThrows
            @Override
            public void onNext(GrpcRequest request) {
                String serviceId = request.getServiceId();
                params = new LinkedHashMap<>(request.getParamsMap());
                String version = request.getVersion();
                byte[] content = request.getContent().toByteArray();

                //get gsp context from filter
                var context = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_CONTEXT);
                Map<String,String> gspContext = (HashMap<String, String>) SerializerFactory.getDeserializer(SerializeType.Json).deserialize(context, HashMap.class);

                if (!gspContext.containsKey(ConstanceVarible.GSP_RPC)
                        || gspContext.get(ConstanceVarible.GSP_RPC) == null
                        || gspContext.get(ConstanceVarible.GSP_RPC).length() == 0
                        || !"true".equalsIgnoreCase(gspContext.get(ConstanceVarible.GSP_RPC))){
                    var state = handleException(Status.PERMISSION_DENIED.withDescription("Only allow invoke in iGIX rpc framework").asException());
                    handleException(state);
                    throw new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT,
                            DefaultExceptionProperties.RESOURCE_FILE,
                            ExceptionErrorCode.illegalRPCCall,
                            new String[]{},
                            null, ExceptionLevel.Error, false);
                }


                //var contextId = gspContext.get("ContextId");
                String serviceUnitName = gspContext.get(ConstanceVarible.GSP_MSU);
                String tenantIdString = gspContext.get(ConstanceVarible.GSP_RPC_TENANT);
                Integer tenantId = !StringUtils.isEmpty(tenantIdString) ? Integer.valueOf(tenantIdString) : null;

                //RpcContext: prepare tenant and msu
                RpcContext rpcContext = RpcContext.getInstance();
                rpcContext.putContext("transmitte-msu",serviceUnitName);
                rpcContext.putContext("transmitte-tenantid",tenantId);
                RpcContextResolver.replay(rpcContext,
                        RpcContextTransmittee.TransmitteType.TENANT.getValue(),
                        RpcContextTransmittee.TransmitteType.MSU.getValue());

                gspContext.remove(ConstanceVarible.GSP_RPC);
                gspContext.remove(ConstanceVarible.GSP_CONTEXT_ID);
                gspContext.remove(ConstanceVarible.GSP_MSU);

                var eventContext = gspContext.get(ConstanceVarible.GSP_RPC_CLIENT_ENVENT);
                eventContextDict = SerializerFactory.getDeserializer(SerializeType.Json).deserialize(eventContext, HashMap.class);
                localDict = new HashMap<String, Object>();
                localDict.put(ConstanceVarible.LOG_MSU, serviceUnitName);
                serverEventBroker.firePreRpcInvokeEvent(eventContextDict, params, localDict);

                try {
                    var intermanager = SpringBeanUtils.getBean(InternalServiceManageService.class);
                    var rpcServiceMethodDefinition = intermanager.getRpcMethodDefinition(serviceId);
                    swapStream = new ByteArrayInputStream(content);
                    params = Utils.handleStreamParam(rpcServiceMethodDefinition,params,swapStream);
                    var parameters = RpcSerializeUtil.deSerializeParameter4RPC2(rpcServiceMethodDefinition,params);
//            var parameters = RpcSerializeUtil.deSerializeParameter(rpcServiceMethodDefinition,params);
                    result = rpcServer.invokeService(serviceId, "", parameters, GspSerializeType.Protobuf).toString();

                } catch (Throwable e) {
                    this.onError(e);
                } finally {
                    RpcContextResolver.restore(rpcContext,RpcContextTransmittee.TransmitteType.TENANT.getValue());
//                    restoreTenant(tenantId);
                    swapStream.close();
                }

                var contextDict = new HashMap<String, String>();
                serverEventBroker.firePostRpcInvokeEvent(contextDict, result, localDict);
                var eventContextStr = SerializerFactory.getSerializer(SerializeType.Json).serializeToString(contextDict);
                RpcThreadCacheHolder.setValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT, eventContextStr);
            }

            @Override
            public void onError(Throwable throwable) {
//                var errerMessage = throwable.getMessage();
                var statusRuntimeException = handleException(throwable);
                responseObserver.onError(statusRuntimeException);
                serverEventBroker.fireExceptionRpcInvokeEvent(eventContextDict, params, localDict, statusRuntimeException);
            }

            @Override
            public void onCompleted() {
                GrpcResponse resp = GrpcResponse.newBuilder()
                        .setMessage(result)
                        .build();
                responseObserver.onNext(resp);
                responseObserver.onCompleted();
            }
        };
    }

    private StatusRuntimeException handleException(Throwable e){
        log.error(e.getMessage(),e);
        /**
         * grpc处理自定义异常的两种方式：1.基于metadata处理，2.基于google.rpc.Status方式
         * 方式2简单，但是目前不是gRPC的官方方式，针对所有语言不具有通用性，这里优先使用方式1
         * 参考连接：
         * https://javakk.com/2558.html#title-0
         * https://cloud.tencent.com/developer/article/1618969
         * */
        Metadata metadata = null;
        if(e instanceof CAFRuntimeException && ((CAFRuntimeException)e).isBizException()){
            metadata = GrpcExceptionMetaData.builder()
                    .exceptionType(CAFRuntimeException.class.getTypeName())
                    .level(ExceptionLevel.Error.ordinal())
                    .date(OffsetDateTime.now())
                    .code(((CAFRuntimeException) e).getExceptionCode())
                    .message(e.getMessage())
                    .isBizException(true)
                    .detail(ExceptionUtils.getStackTrace(e))
                    .requestId(UUID.randomUUID().toString())
                    .extensionMessage(((CAFRuntimeException) e).getExtensionMessage())
                    .build()
                    .buildMetaData();

        }else{
            metadata = GrpcExceptionMetaData.builder()
                    .code(ExceptionErrorCode.rpcError)
                    .exceptionType(e.getClass().getTypeName())
                    .level(ExceptionLevel.Error.ordinal())
                    .date(OffsetDateTime.now())
                    .message(e.getMessage())
                    .detail(ExceptionUtils.getStackTrace(e))
                    .requestId(UUID.randomUUID().toString())
                    .build()
                    .buildMetaData();
        }
        var statusRuntimeException =
                Status.UNAVAILABLE.withDescription(e.getMessage()).asRuntimeException(metadata);
        return statusRuntimeException;
    }

}
